package com.keter.framework.core.util;

import com.google.common.util.concurrent.*;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.*;

/**
 * Created by gulixing@msn.com on 2018/3/7.
 */
public class ConcurrentUtil{

    private static final Logger logger = LoggerFactory.getLogger(ConcurrentUtil.class);
    final static int DEFAULT_POOL_SIZE = 10;
//    final static int DEFAULT_SHARE_POOL_SIZE = 20;
    final static String DEFAULT_POOL_NAME = "DEFAULT_WORKER_THREAD";
    /** 线程运行后是否可自动退出：true 退出，false 不退出
     * the JVM shuts down if there are only daemon threads running. */
    final  boolean DEFAULT_DAEMON = true;

    BasicThreadFactory factory;

    private static ConcurrentUtil instance() {
        return new ConcurrentUtil();
    }

//    /** 内置的共享线程池 */
//    private static ListeningExecutorService service = ConcurrentUtil.service(DEFAULT_SHARE_POOL_SIZE);

//    /**
//     * 执行线程，使用默认的共享线程池
//     * @param task
//     */
//    public static void submit(Runnable task){
//        ListenableFuture future = service.submit(task);
//        executeService(service,future);
//    }

    /**
     * 执行线程
     * @param task
     * @param service
     */
    public static void result(Runnable task, ListeningExecutorService service){
        ListenableFuture future = service.submit(task);
        executeService(service,future);
    }

//    public static <T> T submit(ListenableFuture future) throws InterruptedException, ExecutionException {
//        ListenableFuture[] futures = new ListenableFuture[1];
//        futures[0] = future;
//        return (T) submit(service,futures).get(0);
//    }

    /**
     * 执行线程：同步获取结果
     * @param service
     * @param future
     * @param <T>
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    public static <T> T result(ListeningExecutorService service, ListenableFuture future) throws InterruptedException, ExecutionException {
        ListenableFuture[] futures = new ListenableFuture[1];
        futures[0] = future;
        return (T) result(service,futures).get(0);
    }


    /**
     * 执行线程：同步获取结果
     * @param service
     * @param future
     * @param <T>
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    public static <T> List<T> result(ListeningExecutorService service, ListenableFuture... future) throws InterruptedException, ExecutionException {
        final ListenableFuture allFutures = Futures.allAsList(future);
        final ListenableFuture transform = Futures.transformAsync(allFutures, new AsyncFunction<List<T>, Boolean>() {
            @Override
            public ListenableFuture apply(List<T> results){
                return Futures.immediateFuture(results);
            }
        },service);

        Futures.addCallback(transform, new FutureCallback<List>() {
            @Override
            public void onSuccess(List result) {
            }

            @Override
            public void onFailure(Throwable thrown) {
                logger.error("线程执行失败:", thrown);
            }
        },service);
        return  (List<T>) transform.get();
    }

    /**
     * 默认线程池
     * @return
     */
    public static ListeningExecutorService service(){
        return ConcurrentUtil.instance().getService();
    }

    /**
     * 默认线程池
     * @param size 线程池工作线程数
     * @return
     */
    public static ListeningExecutorService service(int size){
        return ConcurrentUtil.instance().getService(size);
    }

    public BasicThreadFactory getFactory(){
        return factory;
    }

    protected ListeningExecutorService getService(){
        return getService(DEFAULT_POOL_SIZE);
    }

    protected ListeningExecutorService getService(int poolSize){
        return MoreExecutors.listeningDecorator(getExecutorService(DEFAULT_POOL_NAME+"["+RandomUtils.nextInt()+"]-%d",poolSize,DEFAULT_DAEMON));
    }

    protected ListeningExecutorService getService(String name, int poolSize, boolean exit){
        return MoreExecutors.listeningDecorator(getExecutorService(name,poolSize,exit));
    }

    private static void executeService(ListeningExecutorService service, ListenableFuture future) {
        Futures.addCallback(future, new FutureCallback() {
            @Override
            public void onSuccess(Object result) {
            }

            @Override
            public void onFailure(Throwable thrown) {
                logger.error("线程执行失败:", thrown);
            }
        },service);
    }

    private  ExecutorService getExecutorService(String name, int poolSize, boolean exit){
        factory = getBasicThreadFactory(name,exit);
        return new ThreadPoolExecutor(
                poolSize,
                poolSize,
                0L,TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                factory,
                new ThreadPoolExecutor.AbortPolicy()
        );
    }

    private  BasicThreadFactory getBasicThreadFactory(String name, boolean exit){
        return new BasicThreadFactory.Builder()
                .namingPattern(name)
                .daemon(exit)
                .build();
    }
}

